package org.mcache;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

/**
 * @Author: tjc
 * @description
 * @Date Created in 14:10 2022-08-21
 */
public class TcpNioServer {

    public static int READ_ABLE = 0x1;
    public static int WRITE_ABLE = 0x2;
    public static int ACCEPTABLE = 0x4;

    private int port;
    private Selector selector;
    private ByteBuffer inputBuffer = ByteBuffer.allocate(256);
    private ByteBuffer outputBuffer = ByteBuffer.allocate(256);

    private List<RedisClient> clients = new LinkedList<>();

    public TcpNioServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) throws IOException {
        int port = 8080;
        TcpNioServer server = new TcpNioServer(port);
        server.open();
    }

    private void open() throws IOException {
        this.selector = openSelector();
        ServerSocketChannel ss = openServerSocketChannel();

        if (selector == null || ss == null) {
            return;
        }
        int port = bindServerSocketChannel(ss);
        if (port == -1) {
            LOG.warning("bind port error");
            return;
        }
        for (; ; ) {
            selector.selectNow();
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            for (Iterator<SelectionKey> itr = selectedKeys.iterator(); itr.hasNext();) {
                SelectionKey key = itr.next();
                itr.remove();
                eventHandle(key);
            }
        }
    }

    private int bindServerSocketChannel(ServerSocketChannel server) {
        while (true) {
            try {
                server.bind(new InetSocketAddress(port));
                server.configureBlocking(false);
                server.register(selector, SelectionKey.OP_ACCEPT);
                return port;
            } catch (IOException e) {
                if (port < 65535) {
                    port++;
                } else {
                    return -1;
                }
            }
        }
    }

    private ServerSocketChannel openServerSocketChannel() {
        ServerSocketChannel tmpServerSocketChannel;
        try {
            tmpServerSocketChannel = ServerSocketChannel.open();
        } catch (IOException e) {
            e.printStackTrace();
            tmpServerSocketChannel = null;
        }
        return tmpServerSocketChannel;
    }

    private Selector openSelector() {
        Selector tmpSelector;
        try {
            tmpSelector = Selector.open();
        } catch (IOException e) {
            LOG.warning("open selector error", e);
            tmpSelector = null;
        }
        return tmpSelector;
    }

    @FunctionalInterface
    interface TcpEventHandler {
        void apply(SelectionKey selectionKey) throws IOException;
    }

    private TcpEventHandler acceptTcpHandler = (key) -> {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        SocketChannel sc;
        try {
            sc = server.accept();
            sc.configureBlocking(false);
            RedisClient client = new RedisClient();
            client.setSc(sc);
            clients.add(client);
            sc.register(selector, SelectionKey.OP_READ, client);
            LOG.debug("accept handler");
        } catch (IOException e) {
            e.printStackTrace();
        }
    };

    private TcpEventHandler readQueryFromClient = (key) -> {
        SocketChannel sc = (SocketChannel) key.channel();
        try {
            inputBuffer.clear();
            sc.read(inputBuffer);
            inputBuffer.flip();
            String commandStr = StandardCharsets.UTF_8.decode(inputBuffer).toString();
            LOG.info("Message read from client: " + commandStr);
            // deal command
            processCommand((RedisClient) key.attachment(), commandStr);
        } catch (IOException e) {
            LOG.warning("client error, close client", e);
            key.cancel();
        }
    };

    private void processCommand(RedisClient client, String commandStr) {

    }


    private TcpEventHandler sendReplyToClient = (key) -> {
//        System.out.println("handle write event");
        SocketChannel client = (SocketChannel) key.channel();
        try {
            outputBuffer.clear();
            outputBuffer.put("server response...".getBytes(StandardCharsets.UTF_8));
            outputBuffer.flip();
            client.write(outputBuffer);
        } catch (Exception e) {
            LOG.warning("client error, close client", e);
            key.cancel();
        }
    };

    private void eventHandle(SelectionKey key) throws IOException {
        if (!key.isValid()) return;
        try {
            if (key.isAcceptable()) {
                acceptTcpHandler.apply(key);
            }
            if (key.isReadable()) {
                readQueryFromClient.apply(key);
            }
            if (key.isWritable()) {
                sendReplyToClient.apply(key);
            }
        } catch (Exception e) {
            e.printStackTrace();
            key.cancel();
        }
    }
}